/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package spendreport;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.walkthrough.common.entity.Alert;
import org.apache.flink.walkthrough.common.entity.Transaction;

/**
 * 欺诈检测器
 * 欺诈检查类 FraudDetector 是 KeyedProcessFunction 接口的一个实现
 * 他的方法 KeyedProcessFunction#processElement 将会在每个交易事件上被调用。
 * 多个事件之间存储信息需要使用到状态，选择使用 KeyedProcessFunction
 * 能够同时提供对状态和时间的细粒度操作，实现更复杂的算法。
 * <p>
 * Skeleton code for implementing a fraud detector.
 */
public class FraudDetector extends KeyedProcessFunction<Long, Transaction, Alert> {

    private static final long serialVersionUID = 1L;
    /**
     * 定义一个小额交易数量为$1
     */
    private static final double SMALL_AMOUNT = 1.00;
    /**
     * 定义一个大金额交易
     */
    private static final double LARGE_AMOUNT = 500.00;
    /**
     * 时间为一分钟内
     */
    private static final long ONE_MINUTE = 60 * 1000;
    /**
     * 使用一个 boolean 型的标记状态来表示是否刚处理过一个小额交易。
     * ValueState能够为被其封装的变量添加容错能力的类型，防止任务失败重启后，状态信息丢失问题
     */
    private transient ValueState<Boolean> flagState;
    /**
     * 设置计时器
     * 当标记状态被设置为 true 时，设置一个在当前时间一分钟后触发的定时器。
     * 当定时器被触发时，重置标记状态。
     * 当标记状态被重置时，删除定时器
     */
    private transient ValueState<Long> timerState;

    /**
     * 用于注册状态
     *
     * @param parameters
     */
    @Override
    public void open(Configuration parameters) {
        /*
          ValueStateDescriptor 包含了 Flink 如何管理变量的一些元数据信息。状态在使用之前需要先被注册。
          状态需要使用 open() 函数来注册状态。
         */
        ValueStateDescriptor<Boolean> flagDescriptor = new ValueStateDescriptor<>(
                "flag",
                Types.BOOLEAN);
        //ValueState 需要使用 ValueStateDescriptor 来创建
        flagState = getRuntimeContext().getState(flagDescriptor);

        ValueStateDescriptor<Long> timerDescriptor = new ValueStateDescriptor<>(
                "timer-state",
                Types.LONG);
        timerState = getRuntimeContext().getState(timerDescriptor);
    }

    @Override
    public void processElement(
            Transaction transaction,
            Context context,
            Collector<Alert> collector) throws Exception {

        // Get the current state for the current key
        // 获取当前状态为当前的key，检查上一次操作是不是为小金额交易
        Boolean lastTransactionWasSmall = flagState.value();

        // Check if the flag is set
        if (lastTransactionWasSmall != null) {
            if (transaction.getAmount() > LARGE_AMOUNT) {
                //Output an alert downstream
                Alert alert = new Alert();
                alert.setId(transaction.getAccountId());
                collector.collect(alert);
            }
            // Clean up our state
            cleanUp(context);
        }

        if (transaction.getAmount() < SMALL_AMOUNT) {
            // set the flag to true
            flagState.update(true);

            long timer = context.timerService().currentProcessingTime() + ONE_MINUTE;
            //设置一分钟定时器
            context.timerService().registerProcessingTimeTimer(timer);
            timerState.update(timer);
        }
    }

    /**
     * 当定时器触发时，将会调用 KeyedProcessFunction#onTimer 方法。
     */
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Alert> out) {
        // remove flag after 1 minute
        timerState.clear();
        flagState.clear();
    }

    private void cleanUp(Context ctx) throws Exception {
        // delete timer
        Long timer = timerState.value();
        ctx.timerService().deleteProcessingTimeTimer(timer);

        // clean up all state
        timerState.clear();
        flagState.clear();
    }
}